package cn.com.pc.content.service;

import cn.com.pc.content.domain.Biz;
import cn.com.pc.content.domain.Site;
import cn.com.pc.content.domain.article.Article;
import cn.com.pc.content.domain.content.ContentType;
import cn.com.pc.content.domain.customizedContent.CustomizedContent;
import cn.com.pc.content.domain.metadata.ContentDataType;
import cn.com.pc.content.domain.metadata.ContentProperty;
import cn.com.pc.content.domain.post.Post;
import cn.com.pc.content.domain.tag.TagHistory;
import cn.com.pc.content.domain.video.Video;
import cn.com.pc.content.dto.PageForm;
import cn.com.pc.content.dto.PageResult;
import cn.com.pc.content.dto.TagHistoryDTO;
import cn.com.pc.content.service.es.ESCurdService;
import cn.com.pc.content.service.es.ElasticEntity;
import cn.com.pc.content.util.ContentUtils;
import cn.com.pc.content.util.DateUtils;
import cn.com.pc.content.util.LocalDateUtils;
import com.mongodb.client.MongoCursor;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.reflect.FieldUtils;
import org.bson.Document;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.domain.Sort;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StopWatch;

import java.lang.reflect.Field;
import java.util.*;
import java.util.concurrent.Future;

@Slf4j
@Service
@RequiredArgsConstructor
public class RebuildEsContentIndexService {

    private final MongoTemplate mongoTemplate;

    private final ESCurdService esCurdService;

    private final MetadataService metadataService;

    private final TagService tagService;

    private final TagHistoryService tagHistoryService;

    private final ThreadPoolTaskExecutor taskExecutor;

    @Value("${rebuildEsIndex.isKafkaConsumerStart:true}")
    private boolean autoStartup;

    @Value("${rebuildEsIndex.pageSize}")
    private int pageSize;

    @Value("${rebuildEsIndex.eachCursorSize}")
    private int eachCursorSize;


    public void rebuild(String indexName) {
        if (autoStartup) {
            throw new RuntimeException("kafka消费者未停止，不进行索引重建");
        }
        //mongodb数据 导出到es
        try {
            StopWatch watch = new StopWatch();
            watch.start("重建内容索引" + indexName);
            exportMongoDbDocToEsIndex(pageSize, taskExecutor, indexName);
            watch.stop();
            log.info( "{}任务已完成，执行时间{}ms", watch.getLastTaskName(), watch.getLastTaskTimeMillis());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 利用游标cursor 遍历mongoDb 数据，批量导入es
     *
     * @param pageSize 每次从mongodb查询数量
     */
    private void exportMongoDbDocToEsIndex(int pageSize, ThreadPoolTaskExecutor executor, String indexName) {
        List<Future<Integer>> futures = new ArrayList<>();
        String[] split = indexName.split("_");
        if (split.length < 3) {
            throw new RuntimeException("indexName有误");
        }
        Site site = Site.getSourceByName(split[1]);
        ContentType contentType = ContentType.getSourceByName(split[2]);
        if (contentType == null) {
            throw new RuntimeException("contentType有误");
        }
        String mongodbCollectionName = ContentUtils.getMongoCollection(contentType);
        Query query = new Query(Criteria.where("site").is(site));
        long documentSize = mongoTemplate.count(query, mongodbCollectionName);

        //根据mongodb数据量和每次连接游标缓存的数据量计算连接次数，避免一次读取过多造成内存溢出
        long connectCursorTime = documentSize % eachCursorSize > 0
                ? (documentSize / eachCursorSize + 1)
                : documentSize / eachCursorSize;
        log.info("mongodbIndex--({}) documentCount:{}", mongodbCollectionName, documentSize);
        log.info("mongodbIndex--({}) connectCursorCount:{}", mongodbCollectionName, connectCursorTime);

        //设置每次查询mongodb的字段，排序
        query.with(Sort.by(new Sort.Order(Sort.Direction.ASC, "_id")));
        setMongoDbQueryField(query, contentType);
        if (query.getFieldsObject().isEmpty()) {
            log.error("mongodb query 无查询属性,结束重建索引");
            return;
        }

        //循环连接mongodb游标，多线程执行遍历数据后导入es
        for (int i = 0; i < connectCursorTime; i++) {
            int cursorCount = i;
            Future<Integer> future = executor.submit(() -> {

                int hasExportCount = cursorCount * eachCursorSize;
                boolean hasInsertToEs = false;
                List<ElasticEntity> elasticEntityList = new ArrayList<ElasticEntity>();

                try (MongoCursor<Document> cursor = mongoTemplate
                        .getCollection(mongodbCollectionName)
                        .find()
                        .projection(query.getFieldsObject())
                        .sort(query.getSortObject())
                        .noCursorTimeout(true)
                        .batchSize(pageSize)
                        .skip(hasExportCount)
                        .limit(eachCursorSize)
                        .cursor()) {

                    while (cursor.hasNext()) {
                        hasExportCount++;
                        hasInsertToEs = false;
                        Document doc = cursor.next();
                        ElasticEntity elasticEntity = getElasticEntity(doc, contentType);
                        elasticEntityList.add(elasticEntity);

                        //每次数据量达到pageSize后插入es
                        if (elasticEntityList.size() == pageSize) {
                            log.info("batch insert to es ,id from {}-{}",
                                    elasticEntityList.get(0).getId(),
                                    elasticEntityList.get(elasticEntityList.size() - 1).getId());
                            esCurdService.insertBatch(indexName, elasticEntityList);
                            elasticEntityList.clear();
                            hasInsertToEs = true;
                        }
                    }
                    //处理最后一批数据数量小于pageSize
                    if (!hasInsertToEs) {
                        log.info("batch insert to es ,id from {}-{}",
                                elasticEntityList.get(0).getId(),
                                elasticEntityList.get(elasticEntityList.size() - 1).getId());
                        esCurdService.insertBatch(indexName, elasticEntityList);
                        elasticEntityList.clear();
                    }

                    log.info("finish export document count: ({})", hasExportCount);
                } catch (Exception e) {
                    log.error("导入数据到es异常,from({})", hasExportCount);
                    e.printStackTrace();
                }
                return cursorCount + 1;
            });
            futures.add(future);
        }

        try {
            log.info("获取mongoDB数据导入ES 索引 执行结果");
            for (Future<Integer> future : futures) {
                log.info("已执行完的cursor:{}", future.get());
            }
            log.info("mongoDB数据导入ES 索引执行完毕");
        } catch (Exception e) {
            log.error("执行过程异常：", e);
        }
    }

    /**
     * 对mongodb查询结果字段处理为es存储实体
     *
     * @param doc mongodb 查询文档
     * @return
     */
    private ElasticEntity getElasticEntity(Document doc, ContentType contentType) {
        ElasticEntity elasticEntity = new ElasticEntity();
        elasticEntity.setId(String.valueOf(doc.get("_id")));
        doc.put("id", String.valueOf(doc.get("_id")));

        if (doc.getDate("createdAt") != null) {
            doc.put("createdAt", DateUtils.toDateString(doc.getDate("createdAt")));
        }
        if (doc.getDate("updatedAt") != null) {
            doc.put("updatedAt", DateUtils.toDateString(doc.getDate("updatedAt")));
        }

        Site site = Site.valueOf(doc.getString("site"));
        Biz biz = Biz.valueOf(doc.getString("biz"));
        //mongodb中无tags标签列表，需要去mysql查询
        if (!CollectionUtils.isEmpty(doc.getList("tagIds", Long.class))) {
            List<Long> tagIds = doc.getList("tagIds", Long.class);
            Set<Long> tagIdSet = new HashSet<Long>(tagIds);
            // todo:优化方案，在tagHistory增加tagNames字段
            Map<String, Object> map = tagService.verifyTagIds(tagIdSet, site);
            doc.put("tagIds", map.get("tagIds"));
            doc.put("tagNames", map.get("tagNames"));
            TagHistory tagHistory = getTagHistory(elasticEntity.getId(), site);
            if (tagHistory != null) {
                doc.put("taggedAt", LocalDateUtils.format(tagHistory.getTagAt()));
                doc.put("taggedBy", tagHistory.getTagBy());
            }
        }

        // 查询需要索引的metaData元数据
        Map<String, Object> metaData = doc.get("metaData", Map.class);
        if (!CollectionUtils.isEmpty(metaData)) {
            for (Map.Entry<String, Object> entry : metaData.entrySet()) {
                ContentProperty contentProperty = metadataService.getContentProperty(entry.getKey(), site, biz, contentType);
                if (contentProperty == null) {continue;}
                if (contentProperty.getHasIndex()) {
                    if (contentProperty.getDataType().equals(ContentDataType.Date)) {
                        metaData.put(entry.getKey(), DateUtils.toDateString((Date) ContentDataType.Date.parse(entry.getValue())));
                    }
                } else {
                    metaData.remove(entry.getKey());
                }
            }
            doc.put("metaData", metaData);
        }

        doc.remove("_id");
        doc.remove("ip");
        doc.remove("risk");
        elasticEntity.setData(doc);
        return elasticEntity;
    }

    private TagHistory getTagHistory(String contentId, Site site) {
        TagHistoryDTO tagHistoryDTO = new TagHistoryDTO();
        tagHistoryDTO.setContentId(contentId);
        tagHistoryDTO.setSite(site);
        cn.com.pc.content.dto.Sort orders = new cn.com.pc.content.dto.Sort();
        orders.setSortBy("tagAt");
        orders.setAsc(false);
        PageResult<TagHistory> tagHistoryPageResult = tagHistoryService.listTagHistoryByDTO(tagHistoryDTO, new PageForm(), orders);
        if (tagHistoryPageResult.getTotal() > 0) {
            return tagHistoryPageResult.getContent().get(0);
        } else {
            return null;
        }
    }

    /**
     * 根据内容实体类字段，设置查询mongodb的对应字段
     *
     * @param query
     */
    private void setMongoDbQueryField(Query query, ContentType contentType) {

        List<Field> allFieldsList = null;
        //todo 待添加ad,product,other 内容实体
        if (contentType == ContentType.Article) {
            allFieldsList = FieldUtils.getAllFieldsList(Article.class);
        } else if (contentType == ContentType.Video) {
            allFieldsList = FieldUtils.getAllFieldsList(Video.class);
        } else if (contentType == ContentType.Post) {
            allFieldsList = FieldUtils.getAllFieldsList(Post.class);
        } else {
            allFieldsList = FieldUtils.getAllFieldsList(CustomizedContent.class);
        }

        query.fields().include("_id");
        allFieldsList.removeIf(field -> "id".equals(field.getName()));
        allFieldsList.forEach(field -> {
            query.fields().include(field.getName());
            log.info("查询mongodb 的字段名：" + field.getName());
        });
    }

}
